/**
 * 
 */
package com.dangdang.task;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import com.alibaba.fastjson.JSON;
import com.dangdang.annotation.KafkaMqConsumer;
import com.dangdang.pojo.ProducePriceChangeEvent;

/**
 *
 * @author ligeng
 * 2019年6月27日
 */
@Component
@KafkaMqConsumer(threads = "3" ,topic = "subQueue")
public class ProductPriceChangeMessageConsumer implements MessageListener<String, String>{
	
	
	@Override
	public void onMessage(ConsumerRecord<String, String> data) {
		String value = data.value();
		System.out.println(value);
		if(StringUtils.isEmpty(value)) {
			return ;
		}
		ProducePriceChangeEvent producePriceChangeEvent = JSON.parseObject(value, ProducePriceChangeEvent.class);
		
		System.out.println(producePriceChangeEvent);
		
		
	}


}
